<?php

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Wire\AMQPTable;
use PhpAmqpLib\Message\AMQPMessage;

require __DIR__ . '/../../vendor/autoload.php';

$config = require(__DIR__ . '/../config.php');
$connection = new AMQPStreamConnection(
    $config['host'],
    $config['port'],
    $config['user'],
    $config['password'],
    $config['vhost']
);

$channel = $connection->channel();


const DEAD_EXCHANGE = 'dead_exchange';
const DEAD_QUEUE = 'dead_queue';
const NORMAL_EXCHANE = 'normal_exchange';
const NORMAL_QUEUE = 'normal_queue';

// 正常交换机和队列
$channel->exchange_declare(NORMAL_EXCHANE, 'direct');


$options = new AMQPTable([
    'x-dead-letter-exchange' => DEAD_EXCHANGE, // 死信交换机
    'x-dead-letter-routing-key' => 'rk_dead', // 死信路由key
]);
$channel->queue_declare(NORMAL_QUEUE, false, false, false, false, false, $options);
$channel->queue_bind(NORMAL_QUEUE, NORMAL_EXCHANE, 'rk_normal');

// 死信交换机和队列
$channel->exchange_declare(DEAD_EXCHANGE, 'direct');
$channel->queue_declare(DEAD_QUEUE, false, false, false, false, false);
$channel->queue_bind(DEAD_QUEUE, DEAD_EXCHANGE, 'rk_dead');

$channel->basic_consume(NORMAL_QUEUE, '', false, false, false, false, function(AMQPMessage $msg) {
    echo $msg->body . PHP_EOL;
    if($msg->body == "test5") {
        echo "reject test5" . PHP_EOL;
        $msg->reject(false);
    } else {
        $msg->ack();
    }
});

while ($channel->is_open()) {
    $channel->wait();
}

$channel->close();
$connection->close();